package com.yf.cassandra.init;

import com.datastax.driver.core.*;
import com.datastax.driver.core.policies.DefaultRetryPolicy;
import com.datastax.driver.core.policies.ExponentialReconnectionPolicy;
import com.yf.cassandra.runner.CapabilityRunnable;
import com.yf.cassandra.utils.ActionTypeEnum;
import com.yf.cassandra.utils.CommonUtil;
import com.yf.cassandra.utils.DateUtil;
import com.yf.cassandra.utils.MonthEnum;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author: yangfeng
 * @Date: 2020/9/10 15:55
 * Email: Feng.Yang@things-matrix.com
 */
@Component
public class InitRunner implements CommandLineRunner, InitializingBean, DisposableBean {

    public static final String keyspace = "device";
    public static final String table = "thingsmatrix_tmx08_report_log";

    public static final String insertSql = "insert into " + keyspace + "." + table + "(" +
            "id," +
            "time," +
            "uid," +
            "sn," +
            "groupId," +
            "modelCode," +
            "modelName," +
            "companyCode," +
            "altitude," +
            "bags," +
            "battery," +
            "BATTERY_ALERT," +
            "cellId," +
            "current," +
            "direction," +
            "distance," +
            "distance_1," +
            "firmware," +
            "gpsLocating," +
            "iccid," +
            "imei," +
            "ipAddress," +
            "lac," +
            "lat," +
            "lng," +
            "mcc," +
            "mnc," +
            "net," +
            "radius," +
            "sensorVersions," +
            "signalStrength," +
            "speed," +
            "temp_f," +
            "voltage) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";

    private Cluster cluster;
    private Session session;

    public static AtomicInteger successCount = new AtomicInteger();
    public static AtomicInteger failedCount = new AtomicInteger();

    @Override
    public void destroy() throws Exception {
        session.close();
        cluster.close();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        PoolingOptions poolingOptions = new PoolingOptions();
        poolingOptions.setCoreConnectionsPerHost(HostDistance.LOCAL, 2);
        poolingOptions.setMaxConnectionsPerHost(HostDistance.LOCAL, 4);
        poolingOptions
                .setMaxRequestsPerConnection(HostDistance.LOCAL, 1024)
                .setMaxRequestsPerConnection(HostDistance.REMOTE, 1500) //默认256
                .setNewConnectionThreshold(HostDistance.LOCAL, 32);

        // addContactPoints:cassandra节点ip withPort:cassandra节点端口 默认9042
        // withCredentials:cassandra用户名密码 如果cassandra.yaml里authenticator：AllowAllAuthenticator 可以不用配置

        cluster = Cluster.builder().addContactPoint("127.0.0.1")
                .withPoolingOptions(poolingOptions)
                .withRetryPolicy(DefaultRetryPolicy.INSTANCE)
                .withReconnectionPolicy(new ExponentialReconnectionPolicy(5000, 5000))
                .build();

        session = cluster.connect();

        // 创建键空间
        String createKeySpaceCQL = "create keyspace if not exists " + keyspace
                + " with replication={'class':'SimpleStrategy', 'replication_factor':2}";
        session.execute(createKeySpaceCQL);

        // 创建列族
        String createTableCQL = "create table if not exists " + keyspace + "." + table + "(" +
                "id text," +
                "time timestamp," +
                "uid uuid," +
                "sn text," +
                "groupId text," +
                "modelCode text," +
                "modelName text," +
                "companyCode text," +
                "altitude int," +
                "bags float," +
                "battery float," +
                "BATTERY_ALERT text," +
                "cellId int," +
                "current float," +
                "direction int," +
                "distance float," +
                "distance_1 float," +
                "firmware text," +
                "gpsLocating int," +
                "iccid text," +
                "imei text," +
                "ipAddress text," +
                "lac int," +
                "lat double," +
                "lng double," +
                "mcc int," +
                "mnc int," +
                "net text," +
                "radius int," +
                "sensorVersions text," +
                "signalStrength int," +
                "speed float," +
                "temp_f float," +
                "voltage float," +
                "PRIMARY KEY (id,time,uid,sn)" +
                ") WITH CLUSTERING ORDER BY (time DESC)";
        session.execute(createTableCQL);
    }

    @Override
    public void run(String... args) throws Exception {
        if (args.length != 3) {
            System.err.println("Params Error!!!! Must input params: dataTotalNum, threadNum, actionType");
            return;
        }

        long totalDataNum = Long.parseLong(args[0]);
        int threadNum = Integer.parseInt(args[1]);
        ActionTypeEnum actionTypeEnum = ActionTypeEnum.valueOf(args[2].toUpperCase());

        // 1.均分总数量，余数舍弃
        long average = CommonUtil.average(totalDataNum, threadNum);
        BigDecimal totalDecimal = BigDecimal.valueOf(average).multiply(BigDecimal.valueOf(threadNum));
        String total = totalDecimal.toPlainString();
        System.out.printf("Total data: %s\n", total);
        System.out.printf("Thread num: %s\n", threadNum);

        // 2.创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(threadNum);

        long start = System.currentTimeMillis();
        System.err.printf("Main Start timestamp: %s \n", start);

        List<CapabilityRunnable> capabilityRunnableList = new ArrayList<>(threadNum);

        String id = "";
        long time = 0l;
        // 3.创建线程
        for (int i = 0; i < threadNum; i++) {
            int factor = i % 3;

            if(factor == 0){
                id = MonthEnum.july.getValue();
                time = DateUtil.randomTimestamp("2020-07-01", "2020-07-31");
            }else if(factor == 1){
                id = MonthEnum.august.getValue();
                time = DateUtil.randomTimestamp("2020-08-01", "2020-08-31");
            }else{
                id = MonthEnum.september.getValue();
                time = DateUtil.randomTimestamp("2020-09-01", DateUtil.format.format(new Date()));
            }

            capabilityRunnableList.add(new CapabilityRunnable(this.session, average, actionTypeEnum, id, time));
        }

        //添加至线程池并启动
        for (CapabilityRunnable oneRunnable : capabilityRunnableList) {
            executorService.execute(oneRunnable);
        }

        executorService.shutdown();
        System.out.println("......");
        // 4.等待结束
        while (!executorService.isTerminated()) {
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        this.session.close();
        this.cluster.close();

        long end = System.currentTimeMillis();

        // 5.输出
        //System.out.printf("****** End timestamp: %s ****** \n", CommonUtil.simpleDateFormat.format(endTime));
        double speed = (end - start) / 1000.0;
        System.out.printf("Speed time: %s \n", speed + "s");

        if (speed == 0) {
            System.exit(0);
        }

        int success = successCount.get();
        int failed = failedCount.get();

        System.out.printf(">>>>>>>>>\nTotal totalSuccess count is : %s\n", success);
        System.out.printf(">>>>>>>>>\nTotal totalSuccess count is : %s\n", failed);

        double tps = new BigDecimal(success).divide(BigDecimal.valueOf(speed), BigDecimal.ROUND_HALF_EVEN).doubleValue();
        System.out.printf("Average TPS: %f \n", tps);
    }
}
